package com.rd.server.modules.sse.service;

import cn.dev33.satoken.stp.StpUtil;
import com.alibaba.fastjson2.JSONObject;
import com.rd.common.enums.MessageType;
import com.rd.server.modules.room.service.RoomUsersService;
import com.rd.server.modules.user.model.po.UserPO;
import com.rd.server.modules.user.service.UserService;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * SSE服务实现类
 * 提供稳定的Server-Sent Events连接管理
 */
@Service
public class SseServiceImpl implements SseService {

    private static final Logger log = LoggerFactory.getLogger(SseServiceImpl.class);
    
    @Resource
    private UserService userService;

    @Resource
    private RoomUsersService roomUsersService;

    /**
     * 存储所有活跃的 SSE 连接
     * key: 用户ID
     * value: 用户的SSE连接对象
     */
    private static final Map<String, SseEmitter> EMITTERS = new ConcurrentHashMap<>();

    /**
     * 存储用户与房间的映射关系
     * key: 用户ID
     * value: 房间ID
     */
    private static final Map<String, String> USER_ROOMS = new ConcurrentHashMap<>();

    /**
     * 存储用户最后心跳时间
     * key: 用户ID
     * value: 最后心跳时间戳
     */
    private static final Map<String, Long> USER_HEARTBEATS = new ConcurrentHashMap<>();

    /**
     * 存储用户连接统计信息
     * key: 用户ID
     * value: 连接统计
     */
    private static final Map<String, ConnectionStats> USER_STATS = new ConcurrentHashMap<>();

    /**
     * 心跳检测任务调度器
     */
    private ScheduledExecutorService heartbeatScheduler;

    /**
     * 心跳超时时间（毫秒）- 60秒
     */
    private static final long HEARTBEAT_TIMEOUT = 60000L;

    /**
     * 心跳检测间隔（毫秒）- 15秒
     */
    private static final long HEARTBEAT_CHECK_INTERVAL = 15000L;

    /**
     * 最大重连次数
     */
    private static final int MAX_RECONNECT_ATTEMPTS = 3;

    /**
     * 连接统计信息
     */
    private static class ConnectionStats {
        private final AtomicLong connectCount = new AtomicLong(0);
        private final AtomicLong disconnectCount = new AtomicLong(0);
        private final AtomicLong errorCount = new AtomicLong(0);
        private volatile long lastConnectTime = 0;
        private volatile long lastDisconnectTime = 0;

        public void onConnect() {
            connectCount.incrementAndGet();
            lastConnectTime = System.currentTimeMillis();
        }

        public void onDisconnect() {
            disconnectCount.incrementAndGet();
            lastDisconnectTime = System.currentTimeMillis();
        }

        public void onError() {
            errorCount.incrementAndGet();
        }

        public long getConnectCount() { return connectCount.get(); }
        public long getDisconnectCount() { return disconnectCount.get(); }
        public long getErrorCount() { return errorCount.get(); }
        public long getLastConnectTime() { return lastConnectTime; }
        public long getLastDisconnectTime() { return lastDisconnectTime; }
    }

//    @PostConstruct
//    public void init() {
//        startHeartbeatTask();
//        log.info("SSE服务初始化完成");
//    }
//
//    @PreDestroy
//    public void destroy() {
//        stopHeartbeatTask();
//        log.info("SSE服务销毁完成");
//    }

    @Override
    public SseEmitter connect() {
        String userId = StpUtil.getLoginIdAsString();
        
        // 检查是否已有连接
        SseEmitter existingEmitter = EMITTERS.get(userId);
        if (existingEmitter != null) {
            log.info("用户{}已有SSE连接，先清理旧连接", userId);
            cleanupUserConnection(userId);
        }

        // 创建新的SSE连接，设置更长的超时时间
        SseEmitter emitter = new SseEmitter(300000L); // 5分钟超时
        
        // 获取或创建用户统计信息
        ConnectionStats stats = USER_STATS.computeIfAbsent(userId, k -> new ConnectionStats());
        stats.onConnect();

        // 设置连接超时回调
        emitter.onTimeout(() -> {
            log.info("SSE连接超时，用户ID: {} (连接次数: {})", userId, stats.getConnectCount());
            cleanupUserConnection(userId);
        });

        // 设置完成回调
        emitter.onCompletion(() -> {
            log.info("SSE连接正常完成，用户ID: {} (连接次数: {})", userId, stats.getConnectCount());
            cleanupUserConnection(userId);
        });

        // 设置错误回调 - 优化错误处理
        emitter.onError((throwable) -> {
            stats.onError();
            
            // 根据错误类型进行不同处理
            if (throwable instanceof IOException) {
                IOException ioException = (IOException) throwable;
                if ("Broken pipe".equals(ioException.getMessage()) || 
                    "Connection reset by peer".equals(ioException.getMessage())) {
                    // 这是客户端主动断开，属于正常情况
                    log.info("客户端主动断开SSE连接，用户ID: {} (错误次数: {})", userId, stats.getErrorCount());
                } else {
                    // 其他IO异常
                    log.warn("SSE连接IO异常，用户ID: {} (错误次数: {}): {}", userId, stats.getErrorCount(), ioException.getMessage());
                }
            } else {
                // 非IO异常
                log.error("SSE连接发生未知错误，用户ID: {} (错误次数: {})", userId, stats.getErrorCount(), throwable);
            }
            
            cleanupUserConnection(userId);
        });

        EMITTERS.put(userId, emitter);
        USER_HEARTBEATS.put(userId, System.currentTimeMillis());
        
        try {
            // 发送连接成功消息
            String connectMessage = JSONObject.toJSONString(Map.of(
                "type", "connect",
                "message", "Connected successfully",
                "timestamp", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")),
                "userId", userId
            ));
            emitter.send(SseEmitter.event().name("connect").data(connectMessage));
            log.info("用户{}建立SSE连接成功 (总连接次数: {})", userId, stats.getConnectCount());
        } catch (IOException e) {
            log.error("SSE连接建立失败，用户ID: {}", userId, e);
            cleanupUserConnection(userId);
            emitter.completeWithError(e);
        }
        
        return emitter;
    }

    @Override
    public void joinRoom(String roomId) {
        String userId = StpUtil.getLoginIdAsString();
        USER_ROOMS.put(userId, roomId);
        UserPO userPO = userService.selectByUserId(userId);

        // 加入房间
        roomUsersService.joinRoom(roomId, userPO);
        String message = userPO.getNickname() + "加入了房间";
        broadcastToRoom(roomId, MessageType.SYSTEM.getCode(), JSONObject.toJSONString(message));
    }

    @Override
    public void leaveRoom() {
        String userId = StpUtil.getLoginIdAsString();
        String roomId = USER_ROOMS.remove(userId);
        if (roomId != null) {
            UserPO userPO = userService.selectByUserId(userId);
            String message = userPO.getNickname() + "离开了房间";
            broadcastToRoom(roomId, MessageType.SYSTEM.getCode(), JSONObject.toJSONString(message));
        }
    }

    @Override
    public void sendMessage(String message) {
        String userId = StpUtil.getLoginIdAsString();
        String roomId = USER_ROOMS.get(userId);
        if (roomId != null) {
            broadcastToRoom(roomId, MessageType.TRANSFER.getCode(), message);
        }
    }

    @Override
    public void sendHeartbeat() {
        String userId = StpUtil.getLoginIdAsString();
        Long lastHeartbeat = USER_HEARTBEATS.get(userId);
        
        if (lastHeartbeat != null) {
            // 更新心跳时间
            USER_HEARTBEATS.put(userId, System.currentTimeMillis());
            log.debug("用户{}发送心跳包", userId);
            
            // 向客户端发送心跳响应
            SseEmitter emitter = EMITTERS.get(userId);
            if (emitter != null) {
                try {
                    String heartbeatResponse = JSONObject.toJSONString(Map.of(
                        "type", "heartbeat",
                        "timestamp", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")),
                        "status", "ok",
                        "serverTime", System.currentTimeMillis()
                    ));
                    emitter.send(SseEmitter.event().name("heartbeat").data(heartbeatResponse));
                } catch (IOException e) {
                    log.warn("向用户{}发送心跳响应失败: {}", userId, e.getMessage());
                    cleanupUserConnection(userId);
                }
            }
        } else {
            log.warn("用户{}未建立SSE连接，无法发送心跳", userId);
        }
    }

    @Override
    public void startHeartbeatTask() {
        if (heartbeatScheduler == null || heartbeatScheduler.isShutdown()) {
            heartbeatScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
                Thread thread = new Thread(r, "SSE-Heartbeat-Checker");
                thread.setDaemon(true);
                return thread;
            });
            
            // 定期检查心跳超时
            heartbeatScheduler.scheduleAtFixedRate(this::checkHeartbeatTimeout, 
                HEARTBEAT_CHECK_INTERVAL, HEARTBEAT_CHECK_INTERVAL, TimeUnit.MILLISECONDS);
            
            // 定期发送服务端心跳
            heartbeatScheduler.scheduleAtFixedRate(this::sendServerHeartbeat, 
                30000, 30000, TimeUnit.MILLISECONDS);
            
            log.info("SSE心跳检测任务已启动，检测间隔: {}ms", HEARTBEAT_CHECK_INTERVAL);
        }
    }

    @Override
    public void stopHeartbeatTask() {
        if (heartbeatScheduler != null && !heartbeatScheduler.isShutdown()) {
            heartbeatScheduler.shutdown();
            try {
                if (!heartbeatScheduler.awaitTermination(5, TimeUnit.SECONDS)) {
                    heartbeatScheduler.shutdownNow();
                }
            } catch (InterruptedException e) {
                heartbeatScheduler.shutdownNow();
                Thread.currentThread().interrupt();
            }
            log.info("SSE心跳检测任务已停止");
        }
    }

    /**
     * 检查心跳超时
     * 清理超时的连接
     */
    private void checkHeartbeatTimeout() {
        long currentTime = System.currentTimeMillis();
        USER_HEARTBEATS.forEach((userId, lastHeartbeat) -> {
            if (currentTime - lastHeartbeat > HEARTBEAT_TIMEOUT) {
                log.warn("用户{}心跳超时，清理连接", userId);
                cleanupUserConnection(userId);
            }
        });
    }

    /**
     * 服务端主动发送心跳
     * 检测连接是否还活跃
     */
    private void sendServerHeartbeat() {
        EMITTERS.forEach((userId, emitter) -> {
            try {
                String serverHeartbeat = JSONObject.toJSONString(Map.of(
                    "type", "server_heartbeat",
                    "timestamp", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")),
                    "status", "ping"
                ));
                emitter.send(SseEmitter.event().name("server_heartbeat").data(serverHeartbeat));
            } catch (IOException e) {
                log.debug("服务端心跳检测到用户{}连接已断开: {}", userId, e.getMessage());
                cleanupUserConnection(userId);
            }
        });
    }

    /**
     * 向房间广播消息
     * 私有方法，用于向指定房间内的所有用户发送消息
     */
    private void broadcastToRoom(String roomId, String eventName, String message) {
        USER_ROOMS.forEach((userId, userRoomId) -> {
            if (roomId.equals(userRoomId)) {
                SseEmitter emitter = EMITTERS.get(userId);
                if (emitter != null) {
                    try {
                        emitter.send(SseEmitter.event().name(eventName).data(message));
                    } catch (IOException e) {
                        log.debug("向用户{}广播消息失败，连接已断开: {}", userId, e.getMessage());
                        cleanupUserConnection(userId);
                    }
                } else {
                    log.debug("用户{}的SSE连接不存在，清理房间映射", userId);
                    USER_ROOMS.remove(userId);
                }
            }
        });
    }

    /**
     * 清理用户连接信息
     * 统一清理用户的SSE连接和房间映射
     */
    private void cleanupUserConnection(String userId) {
        try {
            SseEmitter emitter = EMITTERS.remove(userId);
            if (emitter != null) {
                emitter.complete();
            }
            USER_ROOMS.remove(userId);
            USER_HEARTBEATS.remove(userId);
            
            // 更新统计信息
            ConnectionStats stats = USER_STATS.get(userId);
            if (stats != null) {
                stats.onDisconnect();
            }
            
            log.debug("已清理用户{}的连接信息", userId);
        } catch (Exception e) {
            log.error("清理用户{}连接信息时发生错误", userId, e);
        }
    }

    /**
     * 获取连接统计信息
     * 用于监控和调试
     */
    public Map<String, Object> getConnectionStats() {
        Map<String, Object> stats = new ConcurrentHashMap<>();
        stats.put("totalConnections", EMITTERS.size());
        stats.put("totalRooms", USER_ROOMS.size());
        stats.put("userStats", USER_STATS);
        return stats;
    }
}
